package net.zhangyun.reactor;

import net.zhangyun.utils.MyConsole;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;

public class SchedulerMain {
    private static ExecutorService executorService = new ForkJoinPool();
    private static Scheduler scheduler1 = Schedulers.fromExecutorService(executorService);
    private static Scheduler scheduler2 = Schedulers.parallel();

    public static void main(String[] args) throws InterruptedException {
        publishOnSample();
        MyConsole.printMessage("=========");
        subscribeOnSample();
    }

    private static void publishOnSample() throws InterruptedException {
        Flux.range(1,3)
                .publishOn(scheduler1)  // Take effect for following calls
                .map(number -> {
                    MyConsole.printSingle(number+1);
                    return number + 1;
                })
                .publishOn(scheduler2)  // Take effect for following calls
                .subscribe(MyConsole::printSingle);

        Thread.sleep(1000);
    }

    private static void subscribeOnSample() throws InterruptedException {
        Flux.range(1,3)
                .map(number -> {
                    int newNumber = ++ number;
                    MyConsole.printSingle(newNumber);
                    return newNumber;
                })
                .subscribeOn(scheduler1)    // Take effect upon subscription
                .subscribe(MyConsole::printSingle);

        Thread.sleep(1000);
    }
}
